package cn.me.alphamq.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.config.OffsetLoader;
import cn.me.alphamq.broker.config.TopicLoader;
import cn.me.alphamq.broker.core.commitlog.CommitLogAppendHandler;
import cn.me.alphamq.broker.core.consumequeue.ConsumeQueueAppendHandler;
import cn.me.alphamq.broker.core.consumequeue.ConsumeQueueConsumeHandler;
import cn.me.alphamq.common.model.broker.CommitLogTopicModel;
import cn.me.alphamq.broker.model.consumequeue.ConsumeQueueConsumeReq;
import cn.me.alphamq.broker.netty.broker.BrokerServer;
import cn.me.alphamq.broker.store.RecoverManager;
import cn.me.alphamq.common.entity.Message;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * broker 的启动类
 *
 * @author f
 */
public class BrokerStartUp {

    // 预映射
    private static void preMapping() throws IOException {
        // 加载 topic 信息，并启动定时刷新 topic 配置文件的任务
        TopicLoader topicLoader = new TopicLoader();
        topicLoader.loadProperties();
        topicLoader.startRefreshTopicInfoTask();

        // 加载 offset 信息，并启动定时刷新 offset 文件的任务
        OffsetLoader offsetLoader = new OffsetLoader();
        offsetLoader.loadProperties();
        offsetLoader.startRefreshOffsetInfoTask();

        // 将 topic 和 offset 的数据“预映射”到内存中
        CommitLogAppendHandler commitLogAppendHandler = Container.getCommitLogAppendHandler();
        ConsumeQueueAppendHandler consumeQueueAppendHandler = Container.getConsumeQueueAppendHandler();
        List<CommitLogTopicModel> commitLogTopicModelList = Container.getCommitLogTopicModelList();
        for (CommitLogTopicModel commitLogTopicModel : commitLogTopicModelList) {
            String topicName = commitLogTopicModel.getTopic();
            commitLogAppendHandler.prepareMMapLoading(topicName);
            consumeQueueAppendHandler.prepareMMapLoading(topicName);
        }
    }

    // 初始化和 nameserver 的长连接通道
    private static void initNameServerChannel() {
        Container.getNameServerClient().initConnection();
    }

    // 初始化 broker 的 netty 服务器
    private static void initBrokerServer() {
        new BrokerServer(Container.getBrokerProperties().getBrokerPort()).startServer();
    }

    public static void main(String[] args) throws Exception {
        // 预映射
        preMapping();

        // 恢复数据
        RecoverManager.recover();

        // 连接到 nameserver 上
        initNameServerChannel();

        // 初始化 broker 的 netty 服务器
        initBrokerServer();
    }

    // 模拟向 CommitLog 文件中添加消息
    private static void mockProduce() throws Exception {
        CommitLogAppendHandler commitLogAppendHandler = Container.getCommitLogAppendHandler();
        String topicName = "test_topic";
        for (int i = 0; i < 10; i++) {
            Message message = new Message();
            message.setBody(("this is a test content " + (40 + i)).getBytes());
            commitLogAppendHandler.appendMsg(message);
            TimeUnit.MILLISECONDS.sleep(100);
        }
        System.out.println("生产完毕");
    }

    // 模拟多线程同时消费消息
    private static void mockConsume() {
        String topicName = "test_topic";
        String consumerGroup = "test_consume_group";
        startConsumerThread(topicName, consumerGroup, 0);
        startConsumerThread(topicName, consumerGroup, 1);
    }

    // 启动消费者线程
    private static void startConsumerThread(String topicName, String consumerGroup, int queueId) {
        ConsumeQueueConsumeHandler consumeQueueConsumeHandler = Container.getConsumeQueueConsumeHandler();
        new Thread(() -> {
            ConsumeQueueConsumeReq consumeQueueConsumeReq =
                    new ConsumeQueueConsumeReq(topicName, consumerGroup, queueId, 5);
            while (true) {
                List<Message> messageList = consumeQueueConsumeHandler.consume(consumeQueueConsumeReq);
                if (messageList.isEmpty()) {
                    System.out.println("结束");
                    break;
                }
                for (Message message : messageList) {
                    System.out.println("[queue" + queueId + "]:" + new String(message.getBody()));
                    consumeQueueConsumeHandler.ack(topicName, consumerGroup, queueId);
                }
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        }, "queue" + queueId + "-thread").start();
    }

}
